package com.imooc.consumer.consumer_mode;

import com.utils.Const;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/10-2024
 * @Description: 这里用一句话描述这个类的作用
 */
public class Consumer01 {
    public static void main(String[] args) throws Exception {
        //定义消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer");
        consumer.setNamesrvAddr(Const.NAME_SERVER_SINGLE);

        //消费模式 集群模式
        //集群消费，消费进度是保存在 broker 服务端的
        //consumer.setMessageModel(MessageModel.CLUSTERING);

        /**
         * 消费模式 广播模式
         * 广播模式，消费进度是保存在 客户端的
         * 广播消费是有一个问题的？
         *      假如 消费者 consumer 在 0:00~ 3:00 之间挂了。
         *      但是在该期间，broker 是有收到消息的。
         *      那么在 3:10 consumer 启动之后，在它挂掉的期间，这一段时间的消息是消费不了的。
         */
        consumer.setMessageModel(MessageModel.BROADCASTING);

        //订阅主题
        consumer.subscribe("test_consumer_mode1_topic","*");
        //设置 messageListener
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //启动 消费者
        consumer.start();
        System.out.println("消费者启动成功!!!");
    }
}